package com.study.flink.java.day05_state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

/**
 * 状态后端之RocksDBStateBackend
 */
public class RocksDbStateBackEnd {


    public static void main(String[] args) throws Exception {

        final Logger LOG = LoggerFactory.getLogger(RocksDbStateBackEnd.class);

        // 本地跑需要执行这个
        System.setProperty("HADOOP_USER_NAME", "root");
        // System.setProperty("HADOOP_USER_NAME", "hdfs");

        LOG.debug("-----debug-------{}------------", System.getProperty("java.io.tmpdir"));
        LOG.info("-----info-------{}------------", System.getProperty("java.io.tmpdir"));
        LOG.warn("-----warn------{}------------", System.getProperty("java.io.tmpdir"));
        LOG.error("-----error-------{}------------", System.getProperty("java.io.tmpdir"));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设定状态存储的后端（一定要写全路径）
        // env.setStateBackend(new FsStateBackend("file:///D:\\IDEA\\flink-study\\dir\\day05\\backend"));
        // env.setStateBackend(new FsStateBackend("hdfs://hdp03:8020/flink/state/rocksdb"));
        // 第二个参数开启增量检查点机制(本地测试环境不开)
        RocksDBStateBackend rocks = new RocksDBStateBackend("hdfs://hdp03:8020/flink/state/rocksdb", false);
        // 若是不单独设置，目录在以java.io.tmpdir为key对应的值（路径）
        rocks.setDbStoragePath("file:///D:\\IDEA\\flink-study\\dir\\rocksdb\\flink");
        //rocks.setDbStoragePath("file:///data/rocksdb");
        env.setStateBackend(rocks);

        //设定重启策略，默认策略无限重启，中间states数据保存到taskmanager内存中
        env.enableCheckpointing(10000L); //周期10秒触发checkpointing

        //设定重启策略(重启间隔时间)，确保程序的健壮性
        env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.of(10, TimeUnit.SECONDS)));

        //程序异常退出或者人为cancel掉，不删除checkpointing的数据(默认是删除的CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION)
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // wordcount程序
        DataStream<String> lines = env.socketTextStream(args[0], Integer.parseInt(args[1]));
        SingleOutputStreamOperator<String> lineFlatMap = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                Arrays.stream(line.split("\\W+")).forEach(s -> {
                    out.collect(s);
                });
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lineFlatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                if (s.startsWith("error")) {
                    throw new RuntimeException("error!!!shutdown!!!");
                }
                return Tuple2.of(s.toUpperCase(), 1);
            }
        });
        wordAndOne.keyBy(0).sum(1).print("RocksDbStateBackEnd--->");
        LOG.warn("------------RocksDbStateBackEnd------------");
        env.execute("RocksDbStateBackEnd");
    }

    public static String getTmpdir() {
        // C:\Users\admin\AppData\Local\Temp\
        System.out.println(System.getProperty("java.io.tmpdir"));
        return System.getProperty("java.io.tmpdir");
    }

}
